package com.yc.testThread;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * Created with IntelliJ IDEA.
 * description: 单机分布式事务队列<br><br>
 *     功能：<br>    当一个接口里包含了多个事务时(直连数据库)，使用该对象方法，可以保证多个事务中任意位置报错，所有的事务都回滚，并且可以得到报错堆栈信息<br><br>
 *
 *     使用方法：<br>1. 创建一个TransactionQueue对象<br>
 *              2. 使用TransactionQueue对象.param()生成一个Map对象，用于存servic方法的参数（不能自己创建Map）<br>
 *              3. 把service方法改造成：无返回值，只有一个Map格式的参数<br>
 *              4. 获取方法引用 service对象::方法名<br>
 *              5. 按service顺序使用add(Consumer<Map> f, Map param, Integer overTime)<br>
 *                 把service添加进行TransactionQueue对象【三个参数依次为：方法引用，上一步生成的参数Map，该方法的最大超时时间，单位毫秒，默认1000】<br>
 *              6. 在每个service里返回值前使用静态方法TransactionQueue.startMonitor(param)启动监听（param就是传进去的参数Map）<br>
 *              7. 最后执行transactionQueue.start()启动任务队列<br>
 *              8. 在controller里捕获并处理异常<br>
 *              9. 写finally里调用transactionQueue.clear()清理资源<br>
 *              10. 当任何一个service里的事务报错，controller可以捕获到该异常，所有的service事务都会回滚<br><br>
 *
 *     局限性：<br>  1. 需要改造方法：参数只能传一个Map，不能有返回值（返回值也只能用Map参数传递）<br>
 *  *               2. 只支持单机直连数据库的事务。通过调用网络业务接口的事务问题还是需要用真正的分布式事务框架<br><br>
 * author: yc
 * date: 2021/2/6 11:09
 */
public class TransactionQueue {
    //队列线程标记
    private static final String THREAD_NAME_TAG = "TransactionQueue-";
    //线程名前缀
    private String THREAD_NAME_PREFIX;
    //最大执行时间：5分钟，无论如何都要报超时
    private static final int TOTAL_MAX_TIME = 300000;

    //0 正常运行 1 有节点超时
    private QueueStatus queueStatus = QueueStatus.ACTIVE;
    //当前正在运行节点序号
    private int currentIndex = -1;
    //存在参数里的TransactionQueue对象key名字
    private static final String PARAM_THIS_KEY = "_THIS_";
    //队列长度
    private int queueListSize = 0;

    enum ItemStatus {
        /**
         * 等待中
         */
        WAITING,
        /**
         * 运行中
         */
        RUNNING,
        /**
         * 本节点运行结束
         */
        FINISH,
        /**
         * 本节点及后面的节点都结束
         */
        ALLFINISH
    }
    enum QueueStatus {
        /**
         * 正常
         */
        ACTIVE,
        /**
         * 有节点超时
         */
        OVERTIME,
        /**
         * 有节点报错
         */
        ERROR
    }

    //任务队列
    public List<ItemChildThreadInfo> queueList = new ArrayList<>();

    //创建事务队列时，初始化本事务队列的线程前缀
    public TransactionQueue() {
        THREAD_NAME_PREFIX = THREAD_NAME_TAG + getTag() + "-";
    }

    /**
     * 启动：一个一个任务的执行
     * 前面的线程执行到监控点的位置，才激活下一个线程开始执行
     */
    public void start() {
        if (queueList.isEmpty()) {
            throw new RuntimeException("任务队列里没有任务");
        }
        queueListSize = queueList.size();

        //启动第一个节点
        currentIndex = 0;
        ItemChildThreadInfo firstItem = queueList.get(currentIndex);
        firstItem.setStatus(ItemStatus.RUNNING);
        firstItem.getThread().start();

        //监控第一个节点
        monitNextItem(firstItem);

        //清理资源
        clear();//清理资源
    }

    /**
     * 获取参数（用户要把自己需要的参数传入这个Map中，然后在service里取出来）
     * @return
     */
    public Map param() {
        Map<String, Object> param = new HashMap<>();
        param.put(PARAM_THIS_KEY, this);
        return param;
    }

    /**
     * 通过参数启动监听任务
     * @param param
     */
    public static void startMonitor(Map<String, Object> param) {
        TransactionQueue transactionQueue = (TransactionQueue) param.get(PARAM_THIS_KEY);
        if (null == transactionQueue) {
            throw new RuntimeException("Param中没有key=[" + PARAM_THIS_KEY + "]的TransactionQueue对象");
        }
        //已经从这个中获取TransactionQueue对象，立即解除两者的关系，防止内存溢出
        param.remove(PARAM_THIS_KEY);
        //开始监控
        transactionQueue.monitor();
    }

    /**
     * 监控点(放在方法返回值前 执行该方法)
     */
    private void monitor() {
        //为了防止其中某个异常节点导致后面的节点得不到 "整个任务已经失败"的通知,而继续往下运行
        if (!QueueStatus.ACTIVE.equals(queueStatus)) {
            return;
        }
        //把自己状态改为 已完成（因为监控点在业务代码的最后一行），然后开始监控下一个节点
        int index = getCurrentThreadIndex();
        ItemChildThreadInfo currentItem = queueList.get(index);
        currentItem.setStatus(ItemStatus.FINISH);

        //当自己不是最后一个节点时，开始启动并监控下一个节点
        if (!Integer.valueOf(index).equals(Integer.valueOf(queueListSize - 1))) {
            currentIndex = index + 1;//之所以是当前索引，是因为真正在跑业务程序的是下一个节点
            ItemChildThreadInfo nextItem = queueList.get(currentIndex);
            Thread nextThread = nextItem.getThread();

            //启动下一个节点
            nextItem.setStatus(ItemStatus.RUNNING);
            nextThread.start();
            //监控下一个节点
            monitNextItem(nextItem);
        }
        System.out.println("节点" + index + "及子节点完全结束");
        currentItem.setStatus(ItemStatus.ALLFINISH);
    }

    /**
     * 监控下一个节点
     * @param nextItem
     * @return
     */
    private Integer monitNextItem(ItemChildThreadInfo nextItem) {
        Integer nexRes = null;
        FutureTask<Integer> nextFutureTask = nextItem.getFutureTask();
        try {
            //放宽100毫秒，防止卡在超时时间那一个点上刚好完成，然后还是被报错
            nexRes = nextFutureTask.get(nextItem.getOverTime()+100, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            //判断下一个节点目前的状态
            if (ItemStatus.WAITING.equals(nextItem.getStatus())) {
                queueStatus = QueueStatus.OVERTIME;
                throw new RuntimeException("[任务" + nextItem.getIndex() + "]运行超时");
            }
        } catch (Exception e) {
            queueStatus = QueueStatus.ERROR;
            throw new RuntimeException(e);
        }
        if (nexRes != null) {
            return nexRes;//正常：完全结束
        }
        //正常：自己结束

        try {
            return nextFutureTask.get(TOTAL_MAX_TIME, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            //达到最大运行时间，不再判断状态：哪怕踩点完成也视为超时（由于我们设置的最大超时时间很长，能踩到这个点，那么这个任务已经处于很异常的状态了）
            queueStatus = QueueStatus.OVERTIME;
            throw new RuntimeException("[任务" + nextItem.getIndex() + "]运行超时");
        } catch (Exception e) {
            queueStatus = QueueStatus.ERROR;
            throw new RuntimeException(e);
        }

    }


    /**
     * 清理资源
     */
    public void clear() {
        //把queueList清空
        queueList.clear();
    }

    /**
     * 找当前线程的序号
     * @return
     */
    private int getCurrentThreadIndex() {
        String threadName = Thread.currentThread().getName();
        if (threadName.indexOf(THREAD_NAME_PREFIX) == -1) {
            throw new RuntimeException("当前线程不属于本事务队列");
        }
        return Integer.parseInt(threadName.substring(THREAD_NAME_PREFIX.length()));
    }

    /**
     * 获取10位uuid
     * @return
     */
    private String getTag() {
        String uuid = UUID.randomUUID().toString();
        uuid = uuid.replace("-", "");
        return uuid.substring(22);
    }

    /**
     * 加入队列
     * @param f   无返回值，只有一个Map参数的函数
     * @param param     参数
     * @param overTime 超时时间 传Null默认超时时间为1秒
     * @return
     */
    public TransactionQueue add(Consumer<Map> f, Map param, Integer overTime) {
        overTime = (overTime==null || overTime <= 0)?1000:overTime;
        ItemCallable ic = new ItemCallable(f, param);
        FutureTask ft = new FutureTask(ic);
        Thread it = new Thread(ft, THREAD_NAME_PREFIX + queueList.size());
        queueList.add(new ItemChildThreadInfo(ft, it, overTime, queueList.size()));
        return this;
    }

    /**
     * 任务节点
     */
    class ItemChildThreadInfo {
        private int index;//任务序列号（从0开始）
        private FutureTask futureTask;
        private Thread thread;
        private int overTime;//超时时间
        private ItemStatus status = ItemStatus.WAITING;

        public ItemChildThreadInfo(FutureTask f, Thread t, int m, int i) {
            futureTask = f;
            thread = t;
            overTime = m;
            index = i;
        }

        public FutureTask getFutureTask() {
            return futureTask;
        }

        public Thread getThread() {
            return thread;
        }

        public int getOverTime() {
            return overTime;
        }

        public ItemStatus getStatus() {
            return status;
        }

        public void setStatus(ItemStatus status) {
            this.status = status;
        }

        public int getIndex() {
            return index;
        }
    }

    class ItemCallable implements Callable<Integer> {
        Map param;
        Consumer<Map> method;

        public ItemCallable(Consumer<Map> f, Map p) {
            param = p;
            method = f;
        }

        @Override
        public Integer call() {
            method.accept(param);
            return 1;
        }
    }
}

